package com.ckw.question.config;

import com.fasterxml.jackson.databind.JsonDeserializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
@Slf4j
public class RabbitMQConfiguration implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    public static final String JUDGE_EXCHANGE = "judge.exchange";           // 判题交换机
    public static final String JUDGE_QUEUE = "judge.queue";                 // 判题队列
    public static final String RESPONSE_QUEUE = "response.queue";           // 判题结果返回队列
    public static final String JUDGE_QUEUE_ROUTING_KEY = "judge.queue.routing.key";     // 判题队列key

    public static final String TEST_EXCHANGE = "test.exchange";             // 判题交换机
    public static final String TEST_QUEUE = "test_queue";                   // 测试队列
    public static final String TEST_QUEUE_ROUTING_KEY = "test.queue.routing.key"; // 测试队列key

    public static final String DEAD_LETTER_EXCHANGE = "dead.letter.exchange"; // 死信交换机
    public static final String DEAD_LETTER_QUEUE = "dead.letter.queue";       // 死信队列
    public static final String DEAD_LETTER_QUEUE_ROUTING_KEY = "dead.letter.queue.routing.key"; // 死信队列key

//    public static final String DELAYED_EXCHANGE = "delayed.exchange";          // 延时交换机
//    public static final String DELAYED_QUEUE = "delayed.queue";                // 延时队列
//    public static final String DELAYED_QUEUE_ROUTING_KEY = "delayed.queue.routing.key"; // 延迟队列key

    @Bean // 判题交换机
    public DirectExchange judgeExchange() {
        return new DirectExchange(JUDGE_EXCHANGE, true, false);
    }

    @Bean // 判题队列
    public Queue judgeQueue() {
        Map<String, Object> map = new HashMap<>();
        map.put("x-max-length", 1000);
//        map.put("x-overflow", "reject-publish"); // 超限拒绝新消息
        map.put("x-overflow", "reject-publish-dlx"); // 拒绝新消息，并将其路由到死信交换器
        map.put("x-message-ttl", 60 * 1000);
        // 添加死信交换器配置以便观察被拒绝/过期的消息
        map.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        map.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_ROUTING_KEY);
        return new Queue(JUDGE_QUEUE, true, false, false, map);
    }

    @Bean // 消息返回队列
    public Queue reponseQueue() {
        Map<String, Object> map = new HashMap<>();
        map.put("x-max-length", 1000);
//        map.put("x-overflow", "reject-publish"); // 超限拒绝新消息
        map.put("x-overflow", "reject-publish-dlx"); // 拒绝新消息，并将其路由到死信交换器
        map.put("x-message-ttl", 60 * 1000);
        // 添加死信交换器配置以便观察被拒绝/过期的消息
        map.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        map.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_ROUTING_KEY);
        return new Queue(RESPONSE_QUEUE, true, false, false, map);
    }

    @Bean // 判题队列绑定
    public Binding judgeBinding() {
        return BindingBuilder.bind(judgeQueue()).to(judgeExchange()).with(JUDGE_QUEUE_ROUTING_KEY);
    }

    @Bean // 测试交换机
    public DirectExchange testExchange() {
        return new DirectExchange(TEST_EXCHANGE, true, false);
    }

    @Bean // 测试队列
    public Queue TestQueue() {
        Map<String, Object> map = new HashMap<>();
        map.put("x-max-length", 1000);
//        map.put("x-overflow", "reject-publish"); // 超限拒绝新消息
        map.put("x-overflow", "reject-publish-dlx"); // 拒绝新消息，并将其路由到死信交换器
        map.put("x-message-ttl", 60 * 1000);
        // 添加死信交换器配置以便观察被拒绝/过期的消息
        map.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        map.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_ROUTING_KEY);
        return new Queue(TEST_QUEUE, true, false, false, map);
    }

    @Bean // 测试队列绑定
    public Binding testBinding() {
        return BindingBuilder.bind(TestQueue()).to(judgeExchange()).with(TEST_QUEUE_ROUTING_KEY);
    }

    @Bean // 死信交换机
    public DirectExchange dlxExchange() {
        return new DirectExchange(DEAD_LETTER_EXCHANGE, true, false);
    }

    @Bean // 死信队列
    public Queue dlxQueue() {
        return new Queue(DEAD_LETTER_QUEUE, true, false, false, null);
    }

    @Bean // 死信队列绑定
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(DEAD_LETTER_QUEUE_ROUTING_KEY);
    }

//    // 延时交换机
//    @Bean
//    public CustomExchange delayedExchange() {
//        Map<String, Object> map = new HashMap<>();
//        map.put("x-delayed-type", "direct");
//        return new CustomExchange(DELAYED_EXCHANGE, "x-delayed-message", true, false, map);
//    }
//
//    // 延时队列
//    @Bean
//    public Queue delayedQueue() {
//        return new Queue(DELAYED_QUEUE, true, false, false);
//    }
//    // 延时队列绑定
//    @Bean
//    public Binding delayedBinding() {
//        return BindingBuilder.bind(delayedQueue()).to(delayedExchange()).with(DELAYED_QUEUE_ROUTING_KEY).noargs();
//    }

    @Bean
    public Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
        ObjectMapper objectMapper = new ObjectMapper();
        return new Jackson2JsonMessageConverter(objectMapper);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Jackson2JsonMessageConverter jackson2JsonMessageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
        return rabbitTemplate;
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 消息发送到交换机成功或者失败都会调用， ack为true是成功，cause是原因
        log.info("acktrue");
        if(!ack) {
            log.info("cause : " + cause);
        }
    }

    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        // 发送到队列失败会调用这个方法
        log.info("发送失败");
    }

    /**
     * 并发
     * @param connectionFactory
     * @return
     */
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, Jackson2JsonMessageConverter jackson2JsonMessageConverter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jackson2JsonMessageConverter);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 或 MANUAL 根据需求
        factory.setConcurrentConsumers(10); // 设置并发消费者数量
        factory.setMaxConcurrentConsumers(20); // 设置最大并发消费者数量
        factory.setPrefetchCount(10);  // 设置更大的prefetch值
        return factory;
    }
}